package chapter6.source;

import com.mashibing.flinkjava.code.chapter6.StationLog;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.ParallelSourceFunction;

import java.util.Random;

/**
 * Java - Flink 自定义Source ,实现ParalleSourceFunction接口
 * sid: 基站id
 * callOut : 主叫号码
 * callIn : 被叫号码
 * callType : 通话类型，fail busy barring success
 * callTime : 呼叫时间
 * duration : 通话时长
 */

class MyDefinedParallelSource implements ParallelSourceFunction<StationLog> {

    Boolean flag = true;
    //一般在run中要循环产生数据，这里产生基站日志数据
    @Override
    public void run(SourceContext ctx) throws Exception {
        Random random = new Random();
        String [] callTypes = {"fail","success","barring","busy"};
        while(flag){
            String sid = "sid_"+random.nextInt(10);
            String callOut = "1811234"+(random.nextInt(9000)+ 1000);
            String callIn = "1915678"+(random.nextInt(9000)+ 1000);
            String callType = callTypes[random.nextInt(4)];
            long callTime = System.currentTimeMillis();
            Long duration = Long.valueOf(random.nextInt(50));
            ctx.collect(new StationLog(sid,callOut,callIn,callType,callTime,duration));
            Thread.sleep(2000);
        }
    }

    //当取消Flink任务时被调用
    @Override
    public void cancel() {
        flag = false;

    }
}
public class ParallelSourceTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource ds = env.addSource(new MyDefinedParallelSource());
        ds.print();
        env.execute();
    }
}
